/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nutch.protocol;

import java.net.MalformedURLException;
import java.net.URL;
import java.util.Collection;
import java.util.HashSet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.plugin.Extension;
import org.apache.nutch.plugin.ExtensionPoint;
import org.apache.nutch.plugin.PluginRepository;
import org.apache.nutch.plugin.PluginRuntimeException;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.util.ObjectCache;

/**
 * Creates and caches {@link Protocol} plugins. Protocol plugins should define
 * the attribute "protocolName" with the name of the protocol that they
 * implement. Configuration object is used for caching. Cache key is constructed
 * from appending protocol name (eg. http) to constant
 * {@link Protocol#X_POINT_ID}.
 */
public class ProtocolFactory {

	public static final Logger LOG = LoggerFactory
			.getLogger(ProtocolFactory.class);

	private final ExtensionPoint extensionPoint;

	private final Configuration conf;

	public ProtocolFactory(Configuration conf) {
		this.conf = conf;
		this.extensionPoint = PluginRepository.get(conf)
				.getExtensionPoint(Protocol.X_POINT_ID);
		if (this.extensionPoint == null) {
			throw new RuntimeException(
					"x-point " + Protocol.X_POINT_ID + " not found.");
		}
	}

	/**
	 * Returns the appropriate {@link Protocol} implementation for a url.
	 * 
	 * @param urlString
	 *            Url String
	 * @return The appropriate {@link Protocol} implementation for a given
	 *         {@link URL}.
	 * @throws ProtocolNotFound
	 *             when Protocol can not be found for urlString
	 */
	public synchronized Protocol getProtocol(String urlString)
			throws ProtocolNotFound {
		ObjectCache objectCache = ObjectCache.get(conf);
		try {
			URL url = new URL(urlString);
			String protocolName = url.getProtocol();
			if (protocolName == null)
				throw new ProtocolNotFound(urlString);

			String cacheId = Protocol.X_POINT_ID + protocolName;
			Protocol protocol = (Protocol) objectCache.getObject(cacheId);
			if (protocol != null) {
				return protocol;
			}

			Extension extension = findExtension(protocolName);
			if (extension == null) {
				throw new ProtocolNotFound(protocolName);
			}

			protocol = (Protocol) extension.getExtensionInstance();
			objectCache.setObject(cacheId, protocol);
			return protocol;
		} catch (MalformedURLException e) {
			throw new ProtocolNotFound(urlString, e.toString());
		} catch (PluginRuntimeException e) {
			throw new ProtocolNotFound(urlString, e.toString());
		}
	}

	private Extension findExtension(String name) throws PluginRuntimeException {

		Extension[] extensions = this.extensionPoint.getExtensions();

		for (int i = 0; i < extensions.length; i++) {
			Extension extension = extensions[i];
			if (contains(name, extension.getAttribute("protocolName")))
				return extension;
		}
		return null;
	}

	boolean contains(String what, String where) {
		String parts[] = where.split("[, ]");
		for (int i = 0; i < parts.length; i++) {
			if (parts[i].equals(what))
				return true;
		}
		return false;
	}

	public Collection<WebPage.Field> getFields() {
		Collection<WebPage.Field> fields = new HashSet<WebPage.Field>();
		for (Extension extension : this.extensionPoint.getExtensions()) {
			Protocol protocol;
			try {
				protocol = (Protocol) extension.getExtensionInstance();
				Collection<WebPage.Field> pluginFields = protocol.getFields();
				if (pluginFields != null) {
					fields.addAll(pluginFields);
				}
			} catch (PluginRuntimeException e) {
				// ignore
			}
		}
		return fields;
	}

}
